package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

// 使用该类来统一管理内存中的所有数据
public class MemoryDataCenter {
    // 交换机表 交换机名-交换机对象
    // key 是 exchangeName, value 是 Exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    // 队列表 队列名-队列对象
    // key 是 queueName, value 是 MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    // 绑定表  交换机名-队列名-绑定对象
    // 第一个 key 是 exchangeName, 第二个 key 是 queueName, value 是 Binding 对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    // 消息表 消息ID-消息对象
    // key 是 messageId, value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // 队列消息表  队列名-该队列的所有消息对象
    // key 是 queueName, value 是 Message 链表
    private ConcurrentHashMap<String, LinkedList> queueMessageMap = new ConcurrentHashMap<>();

    // 队列未确认消息表 队列名-该队列中未被确认的消息对象
    // 第一个 key 是 queueName, 第二个 key 是 messageId, value 是未确认的 Message 对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // --------------------------------------------------------------------------
    // 封装交换机操作
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter - insertExchange] 交换机添加成功! exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter - removeExchange] 交换机删除成功! exchangeName=" + exchangeName);
    }

    // --------------------------------------------------------------------------
    // 封装队列操作
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter - insertQueue] 队列添加成功! queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter - deleteQueue] 队列删除成功! queueName=" + queueName);
    }

    // --------------------------------------------------------------------------
    // 封装绑定操作
    public void insertBinding(Binding binding) throws MqException {
        // 通过交换机名获取绑定, 如果得到的哈希表为空, 则创建哈希表
        // 法一
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if(bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingMap.put(binding.getExchangeName(), binding);
//        }
        // 法二
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        synchronized (bindingMap) {
            // 绑定关系已存在哈希表, 抛出异常
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter - insertBinding] 绑定已存在! exchangeName:" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            // 绑定关系不存在哈希表, 则添加绑定关系
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter - insertBinding] 绑定添加成功! exchangeName=" + binding.getExchangeName()
            + ", queueName=" + binding.getQueueName());
    }

    // 删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null) {
            // 该交换机没有绑定任何队列, 说明该删除操作就是错误的
            throw new MqException("[MemoryDataCenter - deleteBinding] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter - removeBinding] 绑定删除成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    // 获取绑定, 两个版本:
    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    // 2. 根据 exchangeName 获取所有的 Binding
    // 此处最差的结果是本次查询的时候查到了空, 但查询过程中其他线程进行了数据插入, 下次查询操作, 就可以查询到新的数据了, 所有无需加锁
    public Binding getBinging(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }


    // --------------------------------------------------------------------------
    // 封装消息操作

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter - addMessage] 消息添加成功! messageId=" + message.getMessageId());
    }

    // 根据 id 查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据 id 删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter - removeMessage] 消息删除成功! messageId=" + messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 先根据队列的名字, 找到该队列对应的消息链表
        // c.computeIfAbsent(a, b) : 如果能查找到 a, 就返回 a对应的数据. 查询不到, 就创建一个 b, 添加到 c中, 再返回 b
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // 再把数据加到 messages 里面
        synchronized (messages) {
            // 链表中不可重复插入数据了 (链表不会覆盖, 哈希表 key值重复会覆盖)
            messages.add(message);
        }
        // 再把消息往消息表中插入一下
        // 重复插入会覆盖之前的消息, 由于消息的内容不会改变, 所以重复插入不影响 (basicProperties & body)
        addMessage(message);
        System.out.println("[MemoryDataCenter - sendMessage] 消息被添加到队列中! messageId=" + message.getMessageId());
    }

    // 从队列中获取消息 (获取并删除)
    public Message pollMessage(String queueName) {
        // 根据队列名获取对应的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 如果队列为空, 则返回 null
        if(messages == null) {
            return null;
        }
        // 如果队列不为空, 则返回 0号位置的消息, 并将该位置的消息删除
        synchronized (messages) {
            // 链表中无消息
            if(messages.size() == 0) {
                return null;
            }
            // 链表中有消息
            Message message = messages.remove(0);
            System.out.println("[MemoryDataCenter - poolMessage] 消息从队列中取出! messageId=" + message.getMessageId());
            return message;
        }
    }
    // 获取指定队列中消息的个数
    public int getMassageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 队列为空
        if(messages == null) {
            return 0;
        }
        // 队列非空
        // 无法对空对象加锁 (因此对空对象提前判断)
        synchronized (messages) {
            return messages.size();
        }
    }

    // --------------------------------------------------------------------------
    // 未确认消息处理

    // 添加未确认消息
    public void addMessageWithAck(String queueName, Message message) {
        // 获取消息名对应的 待确认消息队列 (不存在则创建)
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        // 待确认消息队列添加消息
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter - addMessageWithAck] 消息进入待确认队列! messageId=" + message.getMessageId());
    }

    // 删除未确认消息 (消息已经确认了)
    public void removeMessageWithAck(String queueName, String messageId) {
        // 根据消息名获取待确认消息队列
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        // 待确认队列为空则无需删除
        if(messageHashMap == null) {
            return ;
        }
        // 待确认队列不为空 则 根据消息名 删除指定消息
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter - removeMessageWithAck] 消息从待确认队列中删除! messageId=" + messageId);
    }

    // 获取执行的未确认消息
    public Message getMessageWithAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    // --------------------------------------------------------------------------
    // 从硬盘上获取数据, 持久化到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 1. 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWaitAckMap.clear();

        // 2. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }

        // 3. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for(MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }

        // 4. 恢复所有的绑定数据
        // 查询所有的绑定对象
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        // 绑定涉及到: 交换机 - 队列, 因此需要唯一确定交换机和队列
        for(Binding binding : bindings) {
            // 先获取对应交换机下的所有绑定
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            // 再添加到该交换机下, 该队列的绑定
            bindingMap.put(binding.getQueueName(), binding);
        }

        // 5.恢复所有的消息数据
        for(MSGQueue queue : queues) {
            // 根据队列名获取该队列的所有消息
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            // 将该队列对应的消息添加到 队列-消息表 中
            queueMessageMap.put(queue.getName(), messages);
            // 将每一个消息添加到消息列表中
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }

        // "未确认消息" 不需要恢复
        // 当等待 ack 的过程中, 服务器重启, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" 了
    }

}
